package cn.hyxy.hadoop;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Demo17_ChainMapperMR extends Configured implements Tool {
	@Override
	public int run(String[] args) throws Exception {
		if (args.length != 2) {
			System.out.println("Usage:in");
			return -1;
		}

		Configuration config = getConf();
		FileSystem fs = FileSystem.get(config);
		Path path = new Path(args[1]);
		if (fs.exists(path)) {
			fs.delete(path, true);
		}
		
		Job job = Job.getInstance(config, "DBOutput");
		job.setJarByClass(getClass());

		//添加多个Mapper
		ChainMapper.addMapper(job,  AMapper.class,LongWritable.class,Text.class, Text.class, LongWritable.class,  config);
		ChainMapper.addMapper(job, BMapper.class, Text.class, LongWritable.class,Text.class, LongWritable.class, config);
		//串联Reducer
		ChainReducer.setReducer(job, CntReducer.class,Text.class, LongWritable.class,Text.class, LongWritable.class, config);
		ChainReducer.addMapper(job, CMapper.class,Text.class, LongWritable.class, LongWritable.class,Text.class, config);
		
		job.setInputFormatClass(TextInputFormat.class);
		TextInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, path);

		return job.waitForCompletion(true) ? 1 : 0;

	}

	public static void main(String[] args) throws Exception {
		int code = ToolRunner.run(new Demo17_ChainMapperMR(), args);
		System.exit(code);
	}

	public static class AMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			String[] strings = value.toString().split("\\s+");
			for (String string : strings) {
				context.write(new Text(string), new LongWritable(1));
			}
		}
	}

	public static class BMapper extends Mapper<Text, LongWritable, Text, LongWritable> {
		@Override
		protected void map(Text key, LongWritable value, Mapper<Text, LongWritable, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			if (key.toString().length() < 5) {
				context.write(key, value);
			}
		}
	}
	public static class CntReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
		 @Override
	        public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
	            long sum = 0;
	            for (LongWritable w : values) {
	                sum += w.get();
	            }
	            context.write(key, new LongWritable(sum));
	        }
	}
	
	public static class CMapper extends Mapper< Text, LongWritable, LongWritable, Text>{
		@Override
		protected void map(Text key, LongWritable value, Mapper<Text, LongWritable, LongWritable, Text>.Context context)
				throws IOException, InterruptedException {
			if (value.get()>=2) {
				context.write(value, key);
			}
		}
	}


}
